#include "epolldispatcher.h"
#include "epollserver.h"
#include "epolllistenserver.h"
#include <time.h>

using namespace std;

static const unsigned int diffsize = 100000;

EpollDispatcher::EpollDispatcher(ThreadPool& threadPool,
        EpollClientFactory& factory,
        unsigned int fdTimeoutSecond)
:m_threadPool(threadPool),
m_factory(factory),
m_stopped(false),
m_timeout(fdTimeoutSecond),
m_maxFd(INVALID_SOCKET),
m_queuesize(0),
m_resizing(false)
{
    m_epollServer = NULL;
    m_epollEventThread = NULL;
    m_queue.resize(diffsize,NULL);
    m_thread.start(*this);
}

EpollDispatcher::~EpollDispatcher()
{
    stop();
    ScopedMutex lock(&m_queueMutex);
    for(vector<FdStat*>::iterator it = m_queue.begin();
            it != m_queue.end(); ++it)
    {
        if (*it == NULL)
            continue;
        if (m_epollServer != NULL)
        {
            if ((*it)->m_fd > 0)
                m_epollServer->closingFd((*it)->m_fd);
        }
        if (m_epollEventThread!= NULL)
        {
            if ((*it)->m_fd > 0)
                m_epollEventThread->closingFd((*it)->m_fd);
        }

        if ((*it)->m_client != NULL )
            delete (*it)->m_client;

        delete *it;
    }

    m_queue.clear();
}

void EpollDispatcher::enqueue(int fd)
{
    // 没必要加锁，不从队列中删除
    if (fd <= 0)
        return;
    
    if (m_queue.size() < (unsigned int)fd)
    {
        ScopedMutex lock(&m_queueMutex);
        m_resizing = true;
        m_queue.resize( fd + diffsize, NULL);
        m_resizing = false;
    }
    
    FdStat* fdst = m_queue[fd];
    if (fdst == NULL)
    {
        fdst = new(nothrow) FdStat();
        if ( fdst == NULL )
            return;
        m_queue[fd] = fdst;
    }

    // 防止一个fd接收多次接收事件
    // 适合于在一次接收事件中，完成所有数据的接收
    // 为空，或者没在运行
    if ( fdst->m_enqnum == 0 
        && (fdst->m_client == NULL
            || !(fdst->m_client != NULL && fdst->m_client->m_running)))
    {
        fdst->m_fd = fd;
        ++fdst->m_enqnum;
        ++m_queuesize;
        m_event.set();
    }
    
    if (m_maxFd < fd)
        m_maxFd = fd;
}

void EpollDispatcher::dequeue(int fd)
{
    if (fd <= 0 )
        return;
    
    if (fd <= m_maxFd)
    {
        FdStat* fdst = m_queue[fd];
        if ( fdst == NULL )
            return;
        fdst->clear();
    }
    
    if (fd == m_maxFd)
        --m_maxFd;

    if (m_queue.size() - m_maxFd > diffsize )
    {
        ScopedMutex lock(&m_queueMutex);
        m_resizing = true;
        m_queue.resize(m_queue.size() - diffsize );
        m_resizing = false;
    }
}

void EpollDispatcher::stop()
{
    if (!m_stopped)
    {
        m_stopped = true;
        m_thread.join();
    }
}

void EpollDispatcher::setEpollServer(EpollServer* server)
{
    m_epollServer = server;
}

void EpollDispatcher::setEpollEventThread(EpollEventThread* thread)
{
    m_epollEventThread = thread;
}

bool EpollDispatcher::checkCloseFd(int fd)
{
    if (m_timeout == 0 || fd < 0)
        return false;
    
    while ( m_resizing )
        continue;
    FdStat* st = m_queue[fd];

    unsigned long curseconds = time(NULL);
    if ( st->m_timeout == 0 )
    {
        st->m_timeout = curseconds;
        return false;
    }

    bool btimeout = (curseconds - st->m_timeout >= m_timeout);
    if (m_epollServer != NULL && btimeout )
    {
        m_epollServer->closingFd(fd);
        if ( st->m_client != NULL )
        {
            st->clear();
            delete st->m_client;
            st->m_client = NULL;
        }
    }
    if (m_epollEventThread != NULL && btimeout )
    {
        m_epollEventThread->closingFd(fd);
        if ( st->m_client != NULL )
        {
            st->clear();
            delete st->m_client;
            st->m_client = NULL;
        }
    }
    
    return btimeout;
}

// 有时，没有请求时，queuesize 也不为零，
void EpollDispatcher::resetQueueSize()
{
    ScopedMutex lock(&m_queueMutex);
    int i = 0;
    int validsize = 0;
    while( m_maxFd > 0 && i <= m_maxFd )
    {
        FdStat* fdst = m_queue[i++];
        if ( fdst == NULL )
            continue;
        if ( fdst->m_fd > 0 
                || (fdst->m_client != NULL && fdst->m_client->m_running))
        {
            ++validsize;
        }
    }

    m_queuesize = validsize;
}

void EpollDispatcher::run()
{
    EpollClientSocket* pClient = NULL;
    int invalidsize = 0;
    while(!m_stopped)
    {
        if ( m_queuesize == 0 )
        {
            // usleep(1) 空闲时，cpu 使用率仍比较高
            m_event.wait(1000);
            continue;
        }
        int i       = 0;
        int fd      = 0;
        invalidsize = 0;
        while( i <= m_maxFd)
        {
            //ScopedMutex lock(&m_queueMutex);
            //不用加锁，太慢了
            if( m_resizing )
                continue;
            FdStat* fdst = m_queue[i];

            if ( fdst == NULL )
            {
                // 发现，大批量突然断开，fd 都为 INVALID_SOCKET时，m_queuesize 不为0
                if ( m_maxFd == invalidsize )
                    resetQueueSize();
                
                ++i;
                ++invalidsize;
                continue;
            }

            fd      = fdst->m_fd;
            pClient = fdst->m_client;
            if ( pClient == NULL)
            {
                pClient = m_factory.createClientSocket(fd);
                fdst->m_client = pClient;
            }

            if (fd < 0 || pClient == NULL || (pClient != NULL && pClient->m_running))
            {
                // 发现，大批量突然断开，fd 都为 INVALID_SOCKET时，m_queuesize 不为0
                if ( m_maxFd == invalidsize )
                    resetQueueSize();
                
                ++i;
                ++invalidsize;
                continue;
            }
           
            if(!m_threadPool.start(*pClient))
            {
                // 下次再处理，不需要++i
                checkCloseFd(fd);
                continue;
            }

            // 业务做完时，仅设为 INVALID_SOCKET
            // 一个 client时，第二次发送时，会重置 INVALID_SOCK 为正常fd
            // dequeue，也是设为INVALID_SOCKET，但会做队列大小的调整
            // 调用在 epollserver 中调整，收发消息时不调整队列大小
            pClient->m_running = true;
            m_queuesize     -= fdst->m_enqnum;
            fdst->m_enqnum  = 0;
            fdst->m_fd      = INVALID_SOCKET;
            fdst->m_timeout = 0;
            ++i;
        }
    }
}









